package com.atguigu.PracticalProject.app.dwd.db;

import com.atguigu.PracticalProject.app.BaseSqlApp;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().init(
                2000,
                2,
                "DwdTradeCartAdd"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //读取ods_db
        readOdsDb(tEnv,"DwdTradeCartAdd");

        //获取cart_info
        Table cartInfo = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['user_id'] user_id," +
                "data['course_id'] course_id," +
                "data['course_name'] course_name," +
                "data['cart_price'] cart_price," +
                "date_format(data['create_time'],'yyyy-MM-dd') date_id1," +
                "data['create_time'] create_time," +
                "date_format(data['update_time'],'yyyy-MM-dd') date_id2," +
                "data['update_time'] update_time," +
                "data['sold'] sold," +
                "ts " +
                "from ods_db " +
                "where `database` = 'edu' " +
                "and `table` = 'cart_info' " +
                "and(" +
                "`type` = 'insert' " +
                "or(`type` = 'update' " +
                "and `old`['sold'] = '0')" +
                ")" +
                "");

        //写出到kafka
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                "id string," +
                "user_id string," +
                "course_id string," +
                "course_name string," +
                "cart_price string," +
                "date_id1 string," +
                "create_time string," +
                "date_id2 string," +
                "update_time string," +
                "sold string," +
                "ts bigint" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_CART_ADD));

        cartInfo.executeInsert("dwd_trade_cart_add");

    }

}
